package com.laycoding.layrpc.handler.client;

import com.laycoding.layrpc.entity.LayRpcMessage;
import com.laycoding.layrpc.entity.LayRpcResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.SynchronousQueue;


@ChannelHandler.Sharable
public class RpcProxyHandler extends ChannelInboundHandlerAdapter {

    private ConcurrentHashMap<String,SynchronousQueue<Object>> queueMap = new ConcurrentHashMap<>();

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {

        if (msg instanceof String){
            return;
        }
        System.out.println(msg.toString());
        if (msg instanceof LayRpcResponse){
            LayRpcResponse message = (LayRpcResponse) msg;
            String requestId = message.getRpcId();
            SynchronousQueue<Object> queue = queueMap.get(requestId);
            queue.put(message);
            queueMap.remove(requestId);
        }
    }

    public SynchronousQueue<Object> sendRequest(LayRpcMessage msg, Channel channel) {
         SynchronousQueue<Object> queue = new SynchronousQueue<>();
         queueMap.put(msg.getRpcId(), queue);
         channel.writeAndFlush(msg);
         return queue;
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("client exception is general");
        ctx.channel().close();
        ctx.close();
    }
}